package fnic.prehand.esagent.ap_measure_new2;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;
import org.codehaus.jackson.JsonEncoding;
import org.codehaus.jackson.JsonFactory;
import org.codehaus.jackson.JsonGenerationException;
import org.codehaus.jackson.JsonGenerator;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.action.update.UpdateRequestBuilder;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.index.query.FilterBuilders;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.script.ScriptService.ScriptType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogram;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogram.Interval;
import org.elasticsearch.search.aggregations.bucket.nested.Nested;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Bucket;
import org.elasticsearch.search.aggregations.bucket.terms.Terms.Order;
import org.elasticsearch.search.aggregations.metrics.sum.Sum;
import org.elasticsearch.search.sort.SortOrder;

import fnic.prehand.common.JacksonFormatService;
import fnic.prehand.esagent.EsAgent;
 
public class APHTTPEsAgent extends EsAgent{ 
	private static final Logger logger = Logger.getLogger(APHTTPEsAgent.class);
	private static final JacksonFormatService jsoninstance = new JacksonFormatService();
	
	private static final String cindex=config.getAp_indexName();
//	private static final String stainfo_history=config.getAptype_stainfo();
	private static final String apinfo_history=config.getAptype_apinfo_his();
	private static final String stainfo_history=config.getAptype_stainfo_his();
	private static final String approperty=config.getAptype_approperty();
	private static final String FIELD_APINFO = "apinfo";
	private static final String FIELD_STAINFO = "stainfo";
	
	private static final String AP_MAC = "apmac";
	private static final String AP_LOGIN = "login_time";
	private static final String AP_LOGOUT = "logout_time";
	private static final String AP_ID = "probeId";
	private static final String AP_ANME = "probeName";
	private static final String AP_LOC = "location";
	private static final String AP_BELONG = "mallName";
	JsonGenerator jsonGenerator = null;
	ByteArrayOutputStream stream = null;

	public APHTTPEsAgent(){
		JsonFactory f = new JsonFactory();
		stream = new ByteArrayOutputStream();
		try {
			jsonGenerator = f.createJsonGenerator(stream, JsonEncoding.UTF8);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			logger.error("APHTTPEsAgent Construct fail:"+e.getMessage());
		}
	}
	
	public boolean dealApProperty(String record){
		UpdateRequestBuilder updateRequest;
		Map<String, Object> headmap=null;
		headmap = (Map<String, Object>)jsoninstance.String2Obj(record, "java.util.Map");
		
		if(headmap == null ||!headmap.containsKey(AP_MAC)){
			return false;
		}

		String apmac = (String) headmap.get(AP_MAC);
		updateRequest = client.prepareUpdate(cindex, approperty, apmac).setDoc(record).setUpsert(record);

		UpdateResponse updatersp;
		try{
			updatersp = updateRequest.execute().actionGet();
		}catch(ElasticsearchException e){
			logger.error("ap property fail:"+e.getMessage());
//			e.printStackTrace();
			return false;
		}
		
	    // 索引名称  
	    String _index = updatersp.getIndex();  
	    // type名称  
	    String _type = updatersp.getType();  
	    // 文档ID  
	    String _id = updatersp.getId();  
	    // 索引版本  
	    long _version = updatersp.getVersion();  
	    
	    logger.info("ap property update succeed, index:" +_index+  
	    		           "  type:" +_type+
	    		           "  id:"   +_id+
	    		           "  version:"+_version);
	    
		return true;
	}
	
	public boolean dealAPLog(String record, boolean login){
		Map<String, Object> headmap=null;
		headmap = (Map<String, Object>)jsoninstance.String2Obj(record, "java.util.Map");
		
		if(headmap == null ||
				!headmap.containsKey(AP_MAC) ||
				!(headmap.containsKey(AP_LOGIN) ||headmap.containsKey(AP_LOGOUT))){
			return false;
		}
		
		boolean updatefg = false;
//		boolean profg = false;
		if(login){
			Map<String ,Object> loginmaps = new HashMap<String ,Object>();
			if(headmap.get(AP_ID)!=null){
				loginmaps.put("ap_id", (Integer)headmap.get(AP_ID));
			}
			if(headmap.get(AP_MAC)!=null){
				loginmaps.put("ap_mac", (String)headmap.get(AP_MAC));
			}
			if(headmap.get(AP_ANME)!=null){
				loginmaps.put("ap_name", (String)headmap.get(AP_ANME));
			}
			if(headmap.get(AP_LOC)!=null){
				loginmaps.put("ap_location", (String)headmap.get(AP_LOC));
			}
			if(headmap.get(AP_BELONG)!=null){
				loginmaps.put("ap_mallname", (String)headmap.get(AP_BELONG));
			}
			updatefg = upsertLogin((String) headmap.get(AP_MAC), (String) headmap.get(AP_LOGIN), loginmaps);
		}else{
			updatefg = updateLogout((String) headmap.get(AP_MAC), (String) headmap.get(AP_LOGOUT));
		}
		
		return updatefg;
	}
	
	
	
	/*====================ES处理接口=====================*/
	//"ap_id":1,
//	"ap_name":"ap_1",
//	"ap_mallname":"万达",
//	"ap_location":"x:1, y:1",
	public static boolean upsertLogin(String apmac, String login, Map<String ,Object> loginmaps){
		UpdateRequestBuilder updateRequest;  
		Map<String ,Object> apinfo_maps = null;
		boolean apinfo_overwrite=false;
		GetResponse getResponse = client.prepareGet(cindex, approperty, apmac).execute().actionGet();
		if(getResponse.isExists()){
			Map<String, Object> apMaps = getResponse.getSourceAsMap();
			if(apMaps!=null&&apMaps.containsKey(FIELD_APINFO)){
				apinfo_overwrite = true;
				apinfo_maps = (Map<String ,Object>)apMaps.get(FIELD_APINFO);
				//只更新login
				apinfo_maps.put("ap_login", login);
			}
		}
		
		/*ap或apinfo不存在*/
		if(!apinfo_overwrite){
			apinfo_maps = new HashMap<String ,Object>();
			apinfo_maps.put("ap_stanum", 0);
			apinfo_maps.put("ap_upflow", 0);
			apinfo_maps.put("ap_downflow", 0);
			apinfo_maps.put("ap_totalflow", 0);
			apinfo_maps.put("ap_login", login);
		}
		
		if(loginmaps==null){
			loginmaps = new HashMap<String ,Object>();
		}
		loginmaps.put(FIELD_APINFO, apinfo_maps);
		
		updateRequest = client.prepareUpdate(cindex, approperty, apmac).setDoc(loginmaps).setUpsert(loginmaps);
		UpdateResponse uprsp;
		try{
			uprsp = updateRequest.execute().actionGet();
		}catch(ElasticsearchException e){
			logger.error("ap login fail:"+e.getMessage());
//			e.printStackTrace();
			return false;
		}
		
	    // 索引名称  
	    String _index = uprsp.getIndex();  
	    // type名称  
	    String _type = uprsp.getType();  
	    // 文档ID  
	    String _id = uprsp.getId();  
	    // 索引版本  
	    long _version = uprsp.getVersion();  
	    
	    logger.info("ap login succeed, index:" +_index+  
	    		           "  type:" +_type+
	    		           "  id:"   +_id+
	    		           "  version:"+_version);
	    
		return true;
	}
	

	
	public static boolean updateLogout(String apmac, String logout){
		UpdateRequestBuilder updateRequest;
		String script = null;
		
		GetResponse getResponse = client.prepareGet(cindex, approperty, apmac).execute().actionGet();
		
		Map<String, Object> propetyMaps = getResponse.getSourceAsMap();
		Map<String, Object> nestedApinfo = null;
		if(!propetyMaps.containsKey(FIELD_APINFO)){
			logger.error("no apinfo["+apmac+"]when logout");
			return false;
		}
		
		nestedApinfo = (Map<String, Object>)propetyMaps.get(FIELD_APINFO);
		nestedApinfo.put("ap_logout", logout);
		String login = (String)nestedApinfo.get("ap_login");
		
		if(login!=null&&logout!=null){
			DateFormat timeformat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			long intrevalSecs = 0;
			try {
				Date logindate = timeformat.parse(login);
				Date logoutdate = timeformat.parse(logout);
				intrevalSecs = (logoutdate.getTime() - logindate.getTime())/1000;
			} catch (ParseException e) {
				logger.error("login/logout dateformat trans error:"+e.getMessage());
			}
			nestedApinfo.put("ap_timelength", intrevalSecs); //以秒为单位
		}
		
		IndexResponse indexResponse = client.prepareIndex(cindex, apinfo_history)  
	            .setSource(nestedApinfo)
	            .setParent(apmac)
	            .execute()  
	            .actionGet();  
		if(indexResponse == null){
			logger.error("apMoveHistory move apmac:"+apmac+" fail");
			return false;
		}
		boolean delstainfo = EsUpdate.staslogoutByAP(apmac);
		if(!delstainfo){
			logger.error("update delete stainfo, move to history fail");
		}
		
		updateRequest = client.prepareUpdate(cindex, approperty, apmac);
		script = "ctx._source.remove(\"apinfo\"); ctx._source.remove(\"stainfo\")";  //删除FIELD_APINFO
		UpdateResponse updatersp;
		try{
			updatersp = updateRequest.setScript(script, ScriptType.INLINE).execute().actionGet();
		}catch(ElasticsearchException e){
			logger.error("update remove apinfo|stainfo fail:"+e.getMessage());
			return false;
		}
		
	    String _index = updatersp.getIndex();  
	    String _type = updatersp.getType();  
	    String _id = updatersp.getId();  
	    long _version = updatersp.getVersion();  
	    logger.info("ap logout succeed, index:" +_index+  
	    		           "  type:" +_type+
	    		           "  id:"   +_id+
	    		           "  version:"+_version);
		return true;
	}
	
	
	public String allAps(String sortField, boolean desc, int from, int size){
//		FilterBuilder serviceAp = FilterBuilders.existsFilter("ap_login");
//		FilterBuilder serviceAp = FilterBuilders.missingFilter("ap_logout").nullValue(true);
//		HasChildFilterBuilder apInfoFilter = FilterBuilders.hasChildFilter(apinfo, serviceAp);
		SearchResponse response = client.prepareSearch(cindex)  
	            .setTypes(approperty)  
	            .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)  
	            .setPostFilter(FilterBuilders.nestedFilter(FIELD_APINFO, FilterBuilders.matchAllFilter()))
	            .addSort(FIELD_APINFO+"."+sortField, !desc?SortOrder.ASC:SortOrder.DESC)
	            .addAggregation(AggregationBuilders.nested("ap_apflow_aggr").path(FIELD_APINFO)
	            		.subAggregation(AggregationBuilders.sum("sum_ap_flow").field("ap_totalflow")))
	            .addAggregation(AggregationBuilders.nested("ap_stanum_aggr").path(FIELD_APINFO)
	            		.subAggregation(AggregationBuilders.sum("sum_sta_num").field("ap_stanum")))
	            .setFrom(from).setSize(size).setExplain(true)  
	            .execute()  
	            .actionGet(); 

	    long all_apnum = response.getHits().getTotalHits();
//	    long all_stanum = stanumByAp(null);
	    
	    Aggregations aggres = response.getAggregations();
	    Nested netstaggr = aggres.get("ap_apflow_aggr");
	    Sum sumflow = netstaggr.getAggregations().get("sum_ap_flow");
	    long all_flow = (long)sumflow.getValue();
	    
	    Aggregations sta_aggres = response.getAggregations();
	    Nested stanumaggr = sta_aggres.get("ap_stanum_aggr");
	    Sum sumstanum = stanumaggr.getAggregations().get("sum_sta_num");
	    long all_stanum = (long)sumstanum.getValue();
	    
	    SearchHits hits = response.getHits(); 
	    List<Map<String, Object>> aplist = new ArrayList<Map<String, Object>>();
	    for (SearchHit hit : hits) {
//	    	System.out.println(hit.getSourceAsString());
//	    	String ap_mac = hit.getId();
	    	Map<String, Object> maps = hit.sourceAsMap();
//	    	maps.put("sta_num", stanumByAp(ap_mac));
//	    	Map<String, Object> properties = getAPPro(ap_mac);
//	    	if(properties!=null){
//	    		maps.putAll(properties);
//	    	}
	    	aplist.add(maps);
	    }
	    
	    Map<String, Object> mapRsp = new HashMap<String, Object>();
	    mapRsp.put("ap_count", all_apnum);
	    mapRsp.put("sta_count", all_stanum);
	    mapRsp.put("flow_count", all_flow);
	    mapRsp.put("ap", aplist);
	    
	    String res = jsoninstance.Obj2String(mapRsp);
//	    logger.info("all ap: "+res);
	    return res;
	}
	
	
	public String stasByAp(String apmac, String sortField, boolean desc, int from, int size){	
/*		FilterBuilder apFilter = FilterBuilders.termFilter("ap_mac", apmac);
		
	    SearchResponse response = client.prepareSearch(cindex)
	             .setTypes(stainfo)  
	            .setSearchType(SearchType.QUERY_AND_FETCH)
	            .setPostFilter(apFilter)
	            .addSort(sortField, !desc?SortOrder.ASC:SortOrder.DESC)
	            .setFrom(from).setSize(size).setExplain(true)  
	            .execute()  
	            .actionGet();
	    
	    SearchHits hits = response.getHits(); 

	    List<Map<String, Object>> stalist = new ArrayList<Map<String, Object>>();
//	    System.out.println("ap["+apmac+"] has "+ hits.totalHits()+" sta(s)");
	    for (SearchHit hit : hits) {
//	    	System.out.println(hit.getSourceAsString());
	    	stalist.add(hit.sourceAsMap());
	    }*/
	    
	    List<Map<String, Object>> stalist = stainfosInAp(apmac);
	    String res = jsoninstance.Obj2String(stalist);
//	    logger.info("ap["+apmac+"] has stas: "+res);
	    return res;
	}
	
	private long stanumByAp(String apmac){
/*		FilterBuilder stafilter;
		if(apmac!=null){
			stafilter = FilterBuilders.termFilter("ap_mac", apmac);
		}else{
			stafilter = FilterBuilders.matchAllFilter();
		}
	    SearchResponse response = client.prepareSearch(cindex)
	             .setTypes(stainfo)  
	            .setSearchType(SearchType.COUNT)
	            .setPostFilter(stafilter)   // Filter  
	            .setExplain(true)  
	            .execute()  
	            .actionGet();
	    
	    SearchHits hits = response.getHits();
	    return hits.getTotalHits();*/
	    
	    return stainfosInAp(apmac).size();
	}
	
	private List<Map<String, Object>> stainfosInAp(String apmac){
		/*当前状态查找sta记录*/
		GetResponse getResponse = client.prepareGet(cindex, approperty, apmac).execute().actionGet();
		Map<String, Object> apMaps = getResponse.getSourceAsMap();
		if(apMaps == null || !apMaps.containsKey(FIELD_STAINFO)){
			logger.error("get stainfos in ap[" +apmac+"]  fail,  stainfo not exist!");
			return null;
		}
		
		return (List<Map<String, Object>>) apMaps.get(FIELD_STAINFO);
	}
	
/*	public void  apAggrgate() throws JsonGenerationException, IOException{
		SearchResponse response = client.prepareSearch(cindex)  
	            .setTypes(approperty)  
	            .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)  
	            .setQuery(QueryBuilders.nestedQuery(FIELD_APINFO, FilterBuilders.matchAllFilter()))
//	            .setPostFilter(FilterBuilders.nestedFilter(FIELD_APINFO, FilterBuilders.matchAllFilter()))
	            .addAggregation(
	            	AggregationBuilders.terms("aggr_l1").field("ap_logical_l1")   //Level 1
	            		.subAggregation(AggregationBuilders.nested("l1_netsted_filter").path(FIELD_APINFO)
	            				.subAggregation(AggregationBuilders.sum("l1_sta_num").field("ap_stanum")))
	            		.subAggregation(AggregationBuilders.terms("aggr_l2").field("ap_logical_l2")   //Level 2
	    	            		.subAggregation(AggregationBuilders.nested("l2_netsted_filter").path(FIELD_APINFO)
	    	            				.subAggregation(AggregationBuilders.sum("l2_sta_num").field("ap_stanum")))
	    	            		.subAggregation(AggregationBuilders.terms("aggr_l3").field("ap_logical_l3")   //Level 3
	    	            				.subAggregation(AggregationBuilders.nested("l3_netsted_filter").path(FIELD_APINFO)
	    	    	            				.subAggregation(AggregationBuilders.sum("l3_sta_num").field("ap_stanum")))
	    	    	            )

	            		)
	            )  
	            .execute()  
	            .actionGet(); 
		
	    long all_apnum = response.getHits().getTotalHits();
	    jsonGenerator.writeStartObject();
	    jsonGenerator.writeNumberField("apnum", all_apnum);
	    jsonGenerator.writeNumberField("stanum", 10000);
	    SearchHits hits = response.getHits(); 
	    for (SearchHit hit : hits) {
	    	System.out.println(hit.getId()+" | "+hit.getSourceAsString());
	    }
	    
	    Aggregations aggres = response.getAggregations();
	    Terms termsBucket = aggres.get("aggr_l1");
	    jsonGenerator.writeArrayFieldStart("level1");
	    Iterator<Bucket> bucketIter = termsBucket.getBuckets().iterator();
	    while(bucketIter.hasNext()){
	    	Bucket bkt = bucketIter.next();
	    	String key = bkt.getKey();

	    	Aggregations subAggr= bkt.getAggregations();
	    	Nested nested1 = subAggr.get("l1_netsted_filter");
	    	long l1_num = nested1.getDocCount();
	    	Sum l1_aggr_sum = nested1.getAggregations().get("l1_sta_num");
	    	double l1_sum = l1_aggr_sum.getValue();
	    	System.out.println("[L1] key:"+key+", num:"+l1_num+", sta:"+l1_sum);
	    	jsonGenerator.writeStartObject();
	    	jsonGenerator.writeStringField("name", key);
	    	jsonGenerator.writeNumberField("num", l1_sum);
	    	
	    	Terms termsBucket2 = subAggr.get("aggr_l2");
		    Iterator<Bucket> bucketIter2 = termsBucket2.getBuckets().iterator();
		    while(bucketIter2.hasNext()){
		    	Bucket bkt2 = bucketIter2.next();
		    	String key2 = bkt2.getKey();

		    	Aggregations subAggr2= bkt2.getAggregations();
		    	Nested netsted2 = subAggr2.get("l2_netsted_filter");
		    	long l2_num = netsted2.getDocCount();
		    	Sum l2_aggr_sum = netsted2.getAggregations().get("l2_sta_num");
		    	double l2_sum = l2_aggr_sum.getValue();
		    	System.out.println("[L2] key:("+key+","+key2+ "), num:"+l2_num+", sta:"+l2_sum);
		    	
		    	
		    	Terms termsBucket3 = subAggr2.get("aggr_l3");
			    Iterator<Bucket> bucketIter3 = termsBucket3.getBuckets().iterator();
			    while(bucketIter3.hasNext()){
			    	Bucket bkt3 = bucketIter3.next();
			    	String key3 = bkt3.getKey();

			    	Aggregations subAggr3= bkt3.getAggregations();
			    	Nested netsted3 = subAggr3.get("l3_netsted_filter");
			    	long l3_num = netsted3.getDocCount();
			    	Sum l3_aggr_sum = netsted3.getAggregations().get("l3_sta_num");
			    	double l3_sum = l3_aggr_sum.getValue();
			    	System.out.println("[L3] key:("+key+","+key2+","+key3+"), num:"+ l3_num+", sta:"+l3_sum);
			    	System.out.println("..........");
			    }
			    System.out.println("----------");
		    }
		    System.out.println("==========");
		    jsonGenerator.writeEndObject(); //end object1
	    }
	    
	    jsonGenerator.writeEndArray(); // end level1
	}*/
	
	private static final String bucketAggr = "bucket_level";
	private static final String nestedAggr = "nested_level";
	private static final String sumAggr = "sum_level";
	
	private static final String fieldLevel = "ap_logical.level";
//	private static final String fieldLevel = "level";

	private static final String fieldStitical = "ap_stanum";
	private static final String rspLevel = "level";
	
	private void parseAggrResult(Aggregations aggres, int curlevel, int maxlevel) throws JsonGenerationException, IOException{
	    if(curlevel>maxlevel){
	    	return;
	    }
		Terms termsBucket = aggres.get(bucketAggr+curlevel);
	    jsonGenerator.writeArrayFieldStart(rspLevel+curlevel);
	    Iterator<Bucket> bucketIter = termsBucket.getBuckets().iterator();
	    while(bucketIter.hasNext()){
	    	Bucket bkt = bucketIter.next();
	    	String key = bkt.getKey();

	    	Aggregations subAggrs= bkt.getAggregations();
	    	Nested nested = subAggrs.get(nestedAggr+curlevel);
	    	long ap_num = nested.getDocCount();
	    	Sum aggr_sum = nested.getAggregations().get(sumAggr+curlevel);
	    	double sta_sum = aggr_sum.getValue();
//	    	System.out.println("[L"+curlevel+"] key:"+key+", num:"+ap_num+", sta:"+sta_sum);
	    	jsonGenerator.writeStartObject();
	    	jsonGenerator.writeStringField("name", key);
	    	jsonGenerator.writeNumberField("num", sta_sum);
	    	
	    	parseAggrResult(subAggrs, curlevel+1, maxlevel);
	    	jsonGenerator.writeEndObject();
	    }
	    jsonGenerator.writeEndArray();
	}
	
	
	private void makeSubAggrRequest(AggregationBuilder aggrReq, int curlevel, int maxlevel){
		if(curlevel>maxlevel){
			return;
		}
		
		AggregationBuilder req = AggregationBuilders.terms(bucketAggr+curlevel).field(fieldLevel+curlevel)   //Level i
				.subAggregation(AggregationBuilders.nested(nestedAggr+curlevel).path(FIELD_APINFO)
						.subAggregation(AggregationBuilders.sum(sumAggr+curlevel).field(fieldStitical)));

		makeSubAggrRequest(req, curlevel+1, maxlevel);
		
		aggrReq.subAggregation(req);
	}
	
	
	public String apinfoLevelAggrgate(int level) throws JsonGenerationException, IOException{
		SearchRequestBuilder searchReq = client.prepareSearch(cindex)  
	        .setTypes(approperty)  
	        .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)  
	        .setQuery(QueryBuilders.nestedQuery(FIELD_APINFO, FilterBuilders.matchAllFilter()));
		
		AggregationBuilder stanumAggr = AggregationBuilders.nested(nestedAggr+0).path(FIELD_APINFO)
				.subAggregation(AggregationBuilders.sum(sumAggr+0).field(fieldStitical));
		AggregationBuilder l1Aggr = AggregationBuilders.terms(bucketAggr+1).field(fieldLevel+1)
						.subAggregation(AggregationBuilders.nested(nestedAggr+1).path(FIELD_APINFO)
								.subAggregation(AggregationBuilders.sum(sumAggr+1).field(fieldStitical))
						);
		
		makeSubAggrRequest(l1Aggr, 2, level);
		searchReq.addAggregation(stanumAggr);
		searchReq.addAggregation(l1Aggr);
		
//		for(int i=1; i<=level; i++){
//			(AggregationBuilders.terms(bucketAggr+i).field(fieldLevel+i)   //Level i
//	            		.subAggregation(AggregationBuilders.nested(nestedAggr+i).path(FIELD_APINFO)
//	            				.subAggregation(AggregationBuilders.sum(sumAggr+i).field(fieldStitical))));
//		}
		

		
		SearchResponse response = searchReq.execute().actionGet(); 
		
		
/*		SearchResponse response = client.prepareSearch(cindex)  
	            .setTypes(approperty)  
	            .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)  
	            .setQuery(QueryBuilders.nestedQuery(FIELD_APINFO, FilterBuilders.matchAllFilter()))
//	            .setPostFilter(FilterBuilders.nestedFilter(FIELD_APINFO, FilterBuilders.matchAllFilter()))
	            .addAggregation(
	            	AggregationBuilders.terms(bucketAggr+1).field(fieldLevel+1)   //Level 1
	            		.subAggregation(AggregationBuilders.nested(nestedAggr+1).path(FIELD_APINFO)
	            				.subAggregation(AggregationBuilders.sum(sumAggr+1).field("ap_stanum")))
	            		.subAggregation(AggregationBuilders.terms(bucketAggr+2).field(fieldLevel+2)   //Level 2
	    	            		.subAggregation(AggregationBuilders.nested(nestedAggr+2).path(FIELD_APINFO)
	    	            				.subAggregation(AggregationBuilders.sum(sumAggr+2).field("ap_stanum")))
	    	            		.subAggregation(AggregationBuilders.terms(bucketAggr+3).field(fieldLevel+3)   //Level 3
	    	            				.subAggregation(AggregationBuilders.nested(nestedAggr+3).path(FIELD_APINFO)
	    	    	            				.subAggregation(AggregationBuilders.sum(sumAggr+3).field("ap_stanum")))
	    	    	            )

	            		)
	            )  
	            .execute()  
	            .actionGet(); */
		
		
	    long all_apnum = response.getHits().getTotalHits();
	    jsonGenerator.writeStartObject();
	    jsonGenerator.writeNumberField("apnum", all_apnum);
	    
	    
	    Aggregations aggres = response.getAggregations();
	    Nested nested = aggres.get(nestedAggr+0);
    	long ap_num = nested.getDocCount();
    	
    	Aggregations subAggres = nested.getAggregations();
    	Sum aggr_sum = subAggres.get(sumAggr+0);
    	double sta_sum = aggr_sum.getValue();
    	
//    	System.out.println("[L0] num:"+ap_num+", sta:"+sta_sum);
    	jsonGenerator.writeNumberField("stanum", sta_sum);
    	
    	
	    parseAggrResult(aggres, 1, level);
	    jsonGenerator.writeEndObject();
	    
		jsonGenerator.flush();
		String jsonres = stream.toString();
		stream.reset();
		return jsonres;
	}
	
	/*
	 * 全局sta的展示
	 */
	public String allStaHisAggr(int level, String value, String orderby, boolean desc, int from, int size) throws JsonGenerationException, IOException{
		QueryBuilder filter;

		if (value == null) {
			filter = QueryBuilders.matchAllQuery();
		} else {
			filter = QueryBuilders.termQuery(fieldLevel + level, value);
		}
		
		Order order;
		if(orderby.equals("timelen")){
			order = Order.aggregation("time_sum", !desc);
		}else if(orderby.equals("totalflow")){
			order = Order.aggregation("flow_sum", !desc);
		}else{
			order = Order.count(!desc);
		}
		
		
		SearchRequestBuilder searchReq = client.prepareSearch(cindex)  
		        .setTypes(stainfo_history)  
		        .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
		        .setQuery(filter)
		        .addAggregation(AggregationBuilders.terms("staHisAggr").field("sta_mac")
		        		.subAggregation(AggregationBuilders.sum("time_sum").field("sta_timelength"))
		        		.subAggregation(AggregationBuilders.sum("flow_sum").field("sta_total"))
		        		.order(order)
		        		.size(from+size));
		
		SearchResponse response = searchReq.execute().actionGet(); 
		
		Aggregations aggres = response.getAggregations();
	    Terms termsBucket = aggres.get("staHisAggr");
	    
	    jsonGenerator.writeStartArray();
	    Iterator<Bucket> bucketIter = termsBucket.getBuckets().iterator();
	    int position = 0;
	    while(bucketIter.hasNext()){
	    	Bucket bkt = bucketIter.next();
	    	if(position++ < from){
	    		continue;
	    	}
	    	String key_stamac = bkt.getKey();
	    	long apnum = bkt.getDocCount();
	    	Aggregations subaggres = bkt.getAggregations();
	    	Sum timesumAggr = subaggres.get("time_sum");
	    	Sum flowsumAggr = subaggres.get("flow_sum");
	    	double timelen =  timesumAggr.getValue();
	    	double totalflow = flowsumAggr.getValue();
	    	jsonGenerator.writeStartObject();
	    	jsonGenerator.writeStringField("stamac", key_stamac);
	    	jsonGenerator.writeNumberField("apnum", apnum);
	    	jsonGenerator.writeNumberField("timelen", timelen);
	    	jsonGenerator.writeNumberField("totalflow", totalflow);
	    	jsonGenerator.writeEndObject();
	    }
	    jsonGenerator.writeEndArray();
	    
		jsonGenerator.flush();
		String jsonres = stream.toString();
		stream.reset();
		return jsonres;
	}
	
	/*
	 * 单个sta的展示
	 */
	public String singleStaHisAggr(String stamac, int level, String value) throws JsonGenerationException, IOException{
		QueryBuilder filter;

		if (value == null) {
			filter = QueryBuilders.termQuery("sta_mac", stamac);
		} else {
			QueryBuilder filter1 = QueryBuilders.termQuery(fieldLevel + level,value);
			QueryBuilder filter2 = QueryBuilders.termQuery("sta_mac", stamac);
			filter = QueryBuilders.boolQuery().must(filter1).must(filter2);
		}
		
		SearchRequestBuilder searchReq = client.prepareSearch(cindex)  
		        .setTypes(stainfo_history)  
		        .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
		        .setQuery(filter)
		        .addFields("ap_mac","sta_login","sta_logout")
		        .addSort("sta_login", SortOrder.ASC)
		        .addAggregation(AggregationBuilders.terms("termAggr").field(fieldLevel+(level+1))
		        		.subAggregation(AggregationBuilders.sum("time_sum").field("sta_timelength"))
		        		.subAggregation(AggregationBuilders.sum("flow_sum").field("sta_total")));
		
		SearchResponse response = searchReq.execute().actionGet(); 
		jsonGenerator.writeStartObject();
		//匹配结果
	    SearchHits hits = response.getHits();
	    jsonGenerator.writeArrayFieldStart("hits");
	    for (SearchHit hit : hits) {
	    	System.out.println(hit.getId()+" | "+hit.getSourceAsString());
	    	jsonGenerator.writeStartObject();
	    	
	    	String apmac = null;
	    	String login = null;
	    	String logout = null;
	    	Map<String, Object> appro = null;
	    	if(hit.field("ap_mac")!=null){
	    		apmac = (String)hit.field("ap_mac").getValue();
	    		appro = EsUpdate.apProperty(apmac);
	    	}
	    	if(hit.field("ap_mac")!=null){
	    		login = (String)hit.field("sta_login").getValue();
	    	}
	    	if(hit.field("ap_mac")!=null){
	    		logout = (String)hit.field("sta_logout").getValue();
	    	}
	    	
	    	jsonGenerator.writeStringField("apmac", apmac);
	    	jsonGenerator.writeStringField("login", login);
	    	jsonGenerator.writeStringField("logout", logout);
	    	jsonGenerator.writeObjectFieldStart("ap_logical");
	    	Iterator<Entry<String, Object>> itor = appro.entrySet().iterator();
	    	while(itor.hasNext()){
	    		Entry<String, Object> item = itor.next();
	    		jsonGenerator.writeStringField(item.getKey(), (String)item.getValue());
	    	}
	    	jsonGenerator.writeEndObject();
	    	jsonGenerator.writeEndObject();
	    }
	    jsonGenerator.writeEndArray();
		
	    //统计结果
		Aggregations aggres = response.getAggregations();
	    Terms termsBucket = aggres.get("termAggr");
	    jsonGenerator.writeArrayFieldStart("statical");
	    Iterator<Bucket> bucketIter = termsBucket.getBuckets().iterator();
	    while(bucketIter.hasNext()){
	    	Bucket bkt = bucketIter.next();
	    	String key_stamac = bkt.getKey();
	    	long apnum = bkt.getDocCount();
	    	Aggregations subaggres = bkt.getAggregations();
	    	Sum timesumAggr = subaggres.get("time_sum");
	    	Sum flowsumAggr = subaggres.get("flow_sum");
	    	double timelen =  timesumAggr.getValue();
	    	double totalflow = flowsumAggr.getValue();
	    	jsonGenerator.writeStartObject();
	    	jsonGenerator.writeStringField("key", key_stamac);
//	    	jsonGenerator.writeNumberField("apnum", apnum);
	    	jsonGenerator.writeNumberField("timelen", timelen);
//	    	jsonGenerator.writeNumberField("totalflow", totalflow);
	    	jsonGenerator.writeEndObject();
	    }
	    jsonGenerator.writeEndArray();
	    jsonGenerator.writeEndObject();
	    
	    
		jsonGenerator.flush();
		String jsonres = stream.toString();
		stream.reset();
		return jsonres; 
	}
	
	/*
	 * 新老客户停留时长分析
	 */
	public String customHisAggr(String querystart, int level, String value) throws JsonGenerationException, IOException{
		DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		long curtime = System.currentTimeMillis();
		Date now = new Date(curtime);
		String querynow = dateFormat.format(now);
		QueryBuilder query;
		
		if(querystart==null){
			long starttime = curtime-1000*60*60*24;
			Date start = new Date(starttime);
			querystart = dateFormat.format(start);
		}
		
		QueryBuilder range_query = QueryBuilders.rangeQuery("sta_login").gte(querystart).lt(querynow); 
		if (value == null) {
			query = range_query;
		} else {
			QueryBuilder level_query = QueryBuilders.termQuery(fieldLevel + level,value);
			query = QueryBuilders.boolQuery().must(range_query).must(level_query);
		}
		
		SearchRequestBuilder searchReq = client.prepareSearch(cindex)  
		        .setTypes(stainfo_history)  
		        .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
		        .setQuery(query)
		        .addAggregation(AggregationBuilders.terms("custom_bucket").field("sta_custom")
		        		.subAggregation(AggregationBuilders.dateHistogram("date_bucket").field("sta_login").interval(Interval.DAY)
		        			.subAggregation(AggregationBuilders.sum("time_sum").field("sta_timelength"))
		        			.subAggregation(AggregationBuilders.terms("sta_bucket").field("sta_mac").size(Integer.MAX_VALUE))));
		
		SearchResponse response = searchReq.execute().actionGet(); 
		long matchnum = response.getHits().getTotalHits();
		System.out.println("match number:"+matchnum);
		
		jsonGenerator.writeStartObject();
		Aggregations aggres = response.getAggregations();
		Terms customBuckets = aggres.get("custom_bucket");
		Iterator<Bucket> cusbucketItor = customBuckets.getBuckets().iterator();
		while(cusbucketItor.hasNext()){
			Bucket custom = cusbucketItor.next();
			jsonGenerator.writeObjectFieldStart(custom.getKey());
			
			Aggregations inneraggres = custom.getAggregations();
			DateHistogram dateBucket = inneraggres.get("date_bucket");
		    
		    Iterator<? extends DateHistogram.Bucket> bucketIter = dateBucket.getBuckets().iterator();
		    int  stanum_sum = 0;
		    double stay_sum = 0;  //每天每个用户的平均停留时长
		    int day = 0;
		    while(bucketIter.hasNext()){
		    	DateHistogram.Bucket bkt = bucketIter.next();
		    	day++;
		    	
		    	String key_date = bkt.getKey();        //当前date
		    	long record_num = bkt.getDocCount();   //用户访问ap个数
		    	
		    	Aggregations subaggres = bkt.getAggregations();
		    	Sum timesumAggr = subaggres.get("time_sum");
		    	double timelen =  timesumAggr.getValue();  //用户总停留时长
		    	
		    	Terms statermAggr = subaggres.get("sta_bucket");
		    	int stanum = statermAggr.getBuckets().size(); //用户个数
		    	
		    	System.out.println("date:"+key_date+", num:"+record_num+", sumtime:"+timelen+", stanum:"+ stanum);
		    	stanum_sum += stanum;
		    	stay_sum += timelen/stanum;

		    }
		    //N天平均个数
		    int num = stanum_sum/day;
		    //N天内的平均停留时长
		    double avgStay = stay_sum/day;
		    
	    	jsonGenerator.writeNumberField("stanum", num);
	    	jsonGenerator.writeNumberField("avgstay", avgStay);
	    	jsonGenerator.writeEndObject();
		}
		
		jsonGenerator.writeEndObject();

	   
		jsonGenerator.flush();
		String jsonres = stream.toString();
		stream.reset();
		return jsonres; 
	}
	
	
	/*
	 * 不同类型用户(认证、失败、潜水)停留时长
	 */
	public String customTypeAggr(int interval, int steps, String querystart, int level, String value) throws JsonGenerationException, IOException{
		DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		long curtime = System.currentTimeMillis();
		Date now = new Date(curtime);
		String querynow = dateFormat.format(now);
		
		int[] interval_count_arr= new int[steps+1];
		QueryBuilder query;
		
		if(querystart==null){
			long starttime = curtime-1000*60*60*24;
			Date start = new Date(starttime);
			querystart = dateFormat.format(start);
		}
		
		QueryBuilder range_query = QueryBuilders.rangeQuery("sta_login").gte(querystart).lt(querynow); 
		if (value == null) {
			query = range_query;
		} else {
			QueryBuilder level_query = QueryBuilders.termQuery(fieldLevel + level,value);
			query = QueryBuilders.boolQuery().must(range_query).must(level_query);
		}
		
		SearchRequestBuilder searchReq = client.prepareSearch(cindex)  
		        .setTypes(stainfo_history)  
		        .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
		        .setQuery(query)
		        .addAggregation(AggregationBuilders.terms("customtype_bucket").field("sta_type")
		        		.subAggregation(AggregationBuilders.dateHistogram("date_bucket").field("sta_login").interval(Interval.DAY)
		        				.subAggregation(AggregationBuilders.terms("sta_bucket").field("sta_mac")
		        						.subAggregation(AggregationBuilders.sum("time_sum").field("sta_timelength")))));
		
		SearchResponse response = searchReq.execute().actionGet(); 
		
		jsonGenerator.writeStartObject();
		Aggregations aggres = response.getAggregations();
		Terms customBuckets = aggres.get("customtype_bucket");
		Iterator<Bucket> cusbucketItor = customBuckets.getBuckets().iterator();
		while(cusbucketItor.hasNext()){
			Bucket custom = cusbucketItor.next();
			jsonGenerator.writeObjectFieldStart(custom.getKey());
			
			Aggregations dateaggres = custom.getAggregations();
			DateHistogram dateBucket = dateaggres.get("date_bucket");
		    Iterator<? extends DateHistogram.Bucket> bucketIter = dateBucket.getBuckets().iterator();
		    
//		    int  stanum_sum = 0;
//		    double staytime_sum = 0;  //每天每个用户的平均停留时长
		    int day = 0;
		    while(bucketIter.hasNext()){
		    	DateHistogram.Bucket dateBkt = bucketIter.next();
		    	day++;
		    	Aggregations staaggr= dateBkt.getAggregations();
				Terms statermAggr = staaggr.get("sta_bucket");
			    Iterator<Bucket> stabucketIter = statermAggr.getBuckets().iterator();
			    while(stabucketIter.hasNext()){
			    	Bucket staBkt = stabucketIter.next();
//			    	String key_stamac = staBkt.getKey();
//			    	long apnum = staBkt.getDocCount();
			    	Aggregations subaggres = staBkt.getAggregations();
			    	Sum timesumAggr = subaggres.get("time_sum");
			    	double timelen =  timesumAggr.getValue();
			    	int step = (int)timelen/interval;
			    	if(step<steps){
			    		interval_count_arr[step]++;
			    	}else{
			    		interval_count_arr[steps]++;
			    	}
			    }
		    }
		    
		    for(int i=0; i<steps+1; i++){
		    	String fieldname;
		    	if(i<steps){
		    		fieldname = i*interval+"~"+(i+1)*interval;
		    	}else{
		    		fieldname = ">"+ i*interval;
		    	}
		    	jsonGenerator.writeNumberField(fieldname, interval_count_arr[i]/day);
		    }
		    
		    jsonGenerator.writeEndObject();
		}
		
		jsonGenerator.writeEndObject();
		
		jsonGenerator.flush();
		String jsonres = stream.toString();
		stream.reset();
		return jsonres;
		
//		Aggregations aggres = response.getAggregations();
//		Terms statermAggr = aggres.get("sta_bucket");
//	    Iterator<Bucket> bucketIter = statermAggr.getBuckets().iterator();
//	    while(bucketIter.hasNext()){
//	    	Bucket bkt = bucketIter.next();
//	    	String key_stamac = bkt.getKey();
//	    	long apnum = bkt.getDocCount();
//	    	Aggregations subaggres = bkt.getAggregations();
//	    	Sum timesumAggr = subaggres.get("time_sum");
//	    	double timelen =  timesumAggr.getValue();
//	    	int step = (int)timelen/interval;
//	    	if(step<steps){
//	    		interval_count_arr[step]++;
//	    	}else{
//	    		interval_count_arr[steps]++;
//	    	}
//	    }
	}
	
	
	
	public static void main(String[] args){
		String login = "{\"probeId\":1233,\"probeName\":\"aptest1\",\"location\":\"jiangsu-1-nanjing-23-jiangning12\",\"apmac\":\"a8.15.4d.0c.00.05\",\"login_time\":\"2014-8-15T12:00:00\"}";
		String logout = "{\"probeId\":1233,\"apmac\":\"a8.15.4d.0c.00.05\",\"logout_time\":\"2014-8-16T12:00:00\"}";
		final String indexName = config.getAp_indexName();
		final String apinfo_typeName = config.getAptype_apinfo();
		final String stainfo_typeName = config.getAptype_stainfo();
		final String aphistory_typeName = config.getAptype_apinfo_his();
		final String approperty_typeName = config.getAptype_approperty();
		
		APHTTPEsAgent apesagent = new APHTTPEsAgent();

//		boolean loginfg = apesagent.dealStaLog(login, true);
//		boolean logoutfg = apesagent.dealStafLog(logout, false);
//		System.out.println("login:"+loginfg+", logout:"+logoutfg);
//		String aps = apesagent.allAps("ap_login", true, 0, 100);
//		String stas = apesagent.stasByAp("a8.15.4d.0c.00.04", "sta_login", true, 0, 100);
//		System.out.println(aps);
//		System.out.println(stas);
		
/*		try {
//			String apinfo = apesagent.apinfoLevelAggrgate(4);
//			String allsta1= apesagent.allStaHisAggr(0, null, "timelen", false, 0, 5);
//			String allsta2= apesagent.allStaHisAggr(0, null, "timelen", false, 1, 5);
//			String allsta3= apesagent.allStaHisAggr(3, "european", "ap_num", true, 0, 3);
//			String allsta4= apesagent.allStaHisAggr(3, "european", "ap_num", true, 2, 3);
//			String singlesta1 = apesagent.singleStaHisAggr("bb.00.00.00.00.02", 0, null);
//			String singlesta2 = apesagent.singleStaHisAggr("bb.00.00.00.00.02", 3, "european");
//			System.out.println(apinfo);
//			System.out.println(allsta1);
//			System.out.println(allsta2);
//			System.out.println(allsta3);
//			System.out.println(allsta4);
//			System.out.println(singlesta1);
//			System.out.println(singlesta2);
			
//			System.out.println(apesagent.customHisAggr(15, 0, null));
			
			System.out.println(apesagent.customTypeAggr(5*60, 10, 15, 0, null));
			
		} catch (JsonGenerationException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}*/
	}
}
